package com.shawn.integeration.business.service.impl;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import cn.hutool.json.JSONUtil;
import com.shawn.integeration.business.enums.BrokerMsgStatusEnum;
import com.shawn.integeration.business.mapper.BrokerMessageLogMapper;
import com.shawn.integeration.business.mapper.OrderMapper;
import com.shawn.integeration.business.model.BrokerMessageLog;
import com.shawn.integeration.business.model.Order;
import com.shawn.integeration.business.service.IOrderService;
import com.shawn.integeration.rabbitmq.constants.RabbitConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;
import java.util.Date;

/**
 * @author shawn
 * @description: 订单服务接口实现
 * @date 2024/1/2
 */
@Slf4j
@Service
public class OrderServiceImpl implements IOrderService, RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {
    @Resource
    private RabbitTemplate rabbitTemplate;
    @Resource
    private OrderMapper orderMapper;
    @Resource
    private BrokerMessageLogMapper messageLogMapper;

    @Override
    @Transactional(rollbackFor = Exception.class, isolation = Isolation.DEFAULT)
    public void createOrder(Order order) {
        Date orderTime = new Date();
        int addrow = orderMapper.insert(order);
        if (addrow <= 0) {
            throw new RuntimeException("订单入库失败");
        }
        log.info("订单入库成功");

        // 插入消息记录表数据
        BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
        // 消息唯一ID
        brokerMessageLog.setMessageId(order.getMessageId());
        // 保存消息整体 转为JSON格式存储入库
        brokerMessageLog.setMessage(JSONUtil.toJsonStr(order));
        // 设置消息状态为0 表示发送中
        brokerMessageLog.setStatus(BrokerMsgStatusEnum.SENDING.getCode());
        // 设置消息未确认超时时间窗口为 一分钟
        brokerMessageLog.setNextRetry(DateUtil.offset(orderTime, DateField.MINUTE, RabbitConstants.ORDER_ACK_TIMEOUT));
        brokerMessageLog.setCreateTime(new Date());
        brokerMessageLog.setUpdateTime(new Date());
        addrow = messageLogMapper.insert(brokerMessageLog);
        if (addrow <= 0) {
            throw new RuntimeException("订单生成失败");
        }
        log.info("rabbitMq消息日志入库成功");
        // 发送消息
        CorrelationData correlationData = new CorrelationData(order.getMessageId());
        rabbitTemplate.convertAndSend(RabbitConstants.ORDER_EXCHANGE, RabbitConstants.ORDER_ROUTE_KEY, JSONUtil.toJsonStr(order), correlationData);
    }

    /**
     * confirm机制只保证消息到达exchange,不保证消息路由到正确quene
     * 如果exchange错误,就会触发confirm机制
     *
     * @param correlationData 消息关联的数据
     * @param ack             消息是否正确到达exchange,true-到达;false-未到达
     * @param s               失败原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        log.info("correlationData:[{}],ack:[{}],cause:[{}]", correlationData, ack, s);
        String messageId = correlationData.getId();
        if (ack) {
            //如果confirm返回成功 则进行更新消息投递日志状态
            BrokerMessageLog brokerMessageLog = new BrokerMessageLog();
            brokerMessageLog.setMessageId(messageId);
            brokerMessageLog.setUpdateTime(new Date());
            brokerMessageLog.setStatus(BrokerMsgStatusEnum.SUCCESS.getCode());
            messageLogMapper.updateById(brokerMessageLog);
            log.info("消息confirm成功,更新DB消息投递记录状态");
        } else {
            log.error("消息confirm失败,可尝试重试 或者补偿等手段");
        }
    }

    /**
     * Return 消息机制用于处理一个不可路由的消息,触发条件有以下两种情况:
     * 1. exchange不存在导致消息不可达
     * 2. routingKey路由不到导致消息不可达
     *
     * @param returnedMessage 返回信息
     */
    @Override
    public void returnedMessage(ReturnedMessage returnedMessage) {
        log.error("路由信息不可达,message:[{}]", returnedMessage);
    }
}
